package http

import (
	"net/http"
	"strings"
	"strconv"
	"gateway/conf"
	"gateway/lib"
	"gateway/types"
	"github.com/funlake/gopkg/utils/log"
	"github.com/funlake/gopkg/jobworker"
	"github.com/valyala/fasthttp"
	"os"
	"time"
	"errors"
	"bytes"
	"github.com/funlake/gopkg/breaker"
	"bufio"
	"fmt"
	"github.com/json-iterator/go"
	"gateway/filters"
	"sync"
)
var (
	dispatcher *jobworker.NonBlockingDispatcher
	globalTransport = &http.Transport{
		DisableKeepAlives : false,
		MaxIdleConnsPerHost : conf.GetConfig().GetEnvInt("connection_per_host"),
		//不要设置太短时间，否则会出现close_wait过多的情况，
		// 因为这边在close_wait状态关停，而没有继续发fin到对方，造成close_wait卡死现象
		//TLSHandshakeTimeout:time.Millisecond * 100,
		//DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
		//
		//	return net.DialTimeout(network,addr,time.Duration(time.Second * 3))
		//},
	}
	//fasthttpClient = & fasthttp.Client{}
	maxWorker int
	queueSize int
	errTimeout = errors.New("服务超时")
	errTempDown = errors.New("服务当前不可用,请稍后重试")
)

type Request struct {
	timeout int
}
type response struct{
	code int
	body string
	err  error
	dur  time.Duration
}
type circuitbreaker struct{
	Timeout int `json:"t"`
	Rate int `json:"r"`
	Window int `json:"w"`
	Min int `json:"m"`
}
func NewRequest()(*Request){
	maxWorker,_ = strconv.Atoi(os.Getenv("GATEWAY_WORKER_SIZE"))
	if maxWorker == 0{
		maxWorker = 150
	}
	queueSize,_ = strconv.Atoi(os.Getenv("GATEWAY_JOBQUEUE_SIZE"))
	if queueSize == 0{
		queueSize = 500
	}
	dispatcher = jobworker.NewNonBlockingDispather(maxWorker,queueSize)
	log.Success("Worker number:%d,Queue size:%d",maxWorker,queueSize)
	return &Request{
		timeout: conf.GetConfig().GetEnvInt("upstream_timeout_second"),
	}

}
func (this *Request)ProxyHttpRequest(proxyContext *types.ProxyContext,ctx *fasthttp.RequestCtx) (flag bool,code int,msg string,dur time.Duration){
	r := this.Do(proxyContext,ctx)
	if  r.err != nil {
		return false , r.code, r.err.Error(),r.dur
	}
	if r.code != 200 {
		sv,_ :=  conf.GetConfig().Get("status").GetString(strconv.Itoa(r.code),"")
		//go aliyunlog.Write("gateway","route_proxy_business_error",aliyunlog.Content{fmt.Sprintf("code:%d,api:%s,result:%s",resp.StatusCode,proxy.Mapkey,utils.ByteToStr(body)),"fail"},true)
		return false,r.code,sv,0
	}
	//go lib.SyncMonitor("access", "method="+proxyStruct.Method+",domain="+proxyStruct.Vhost+",route="+proxyStruct.Mapkey, "success=1")
	return true , 200 , r.body,r.dur
}
func (this *Request) Do(proxyContext *types.ProxyContext,ctx *fasthttp.RequestCtx) (resp response) {
	if flag,code,msg := filters.NewFilterHandler().TriggerPreFilters(proxyContext,ctx);!flag{
		return response{code,"",errors.New(msg),0}
	}
	job := jobworker.NewHttpProxyJob(globalTransport,proxyContext.Req,maxWorker, proxyContext.Mapkey)
	now := time.Now()
	brek := breaker.NewBreaker(proxyContext.Mapkey)
	cbsetting := getCircuitBreaker(proxyContext.Mapkey)
	brek.SetTimemout(cbsetting.Timeout)
	brek.SetRate(cbsetting.Rate)
	brek.SetWindow(cbsetting.Window)
	brek.SetMin(cbsetting.Min)
	wg := sync.WaitGroup{}
	//rt   := make(chan response,1)
	var rt response
	postshow := proxyContext.PostData[:min(len(proxyContext.PostData),1000)]
	wg.Add(1)
	brek.Run(func() {
		//此函数区块不能有阻塞动作
		if !dispatcher.Put(job) {
			//rt <- response{503,"",errors.New("服务繁忙,请求队列已满"),0}
			rt = response{503,"",errors.New("服务繁忙,请求队列已满"),0}
		}else{
		//	now := time.Now()
			r := <- job.GetResChan()
			if r.Error == nil {
				buf := new(bytes.Buffer)
				buf.ReadFrom(r.Response.Body)
				//utils.WrapGo(func() {
				for hk, hv := range r.Response.Header {
					ctx.Response.Header.Set(hk, strings.Join(hv, ";"))
				}
				rt = response{r.Response.StatusCode, buf.String(), nil, r.Dur}
				r.Response.Body.Close()
				//}, "response_set")
			}else{
				log.Error(r.Error.Error())
				rt  = response{conf.REQUEST_SERVICE_TEMPORARY_DOWN,"",r.Error,0}
				log.Error("%s,%s",readHeader(ctx.Request.Header.String()+fmt.Sprintf("Response-Time: %s Error-Code: %d",time.Since(now),conf.REQUEST_SERVICE_TEMPORARY_DOWN)),postshow)
			}
		}
	}, func() {
		log.Success("%s,%s",readHeader(ctx.Request.Header.String()+fmt.Sprintf("Response-Time: %s",time.Since(now))),postshow)
		wg.Done()
	}, func(run bool) {
		errcode := 0
		errins := errors.New("")
		if run {
			errcode = conf.REQUEST_TIMEOUT
			errins = errTimeout
		}else{
			errcode = conf.REQUEST_SERVICE_TEMPORARY_DOWN
			errins = errTempDown

		}
		rt = response{errcode,"",errins,0}
		log.Error("%s,%s",readHeader(ctx.Request.Header.String()+fmt.Sprintf("Response-Time: %s Error-Code: %d",time.Since(now),errcode)),postshow)
		wg.Done()
	})
	//estart := time.Now()
	wg.Wait()
	if flag,code,msg := filters.NewFilterHandler().TriggerPostFilters(proxyContext,ctx);!flag{
		return response{code,"",errors.New(msg),0}
	}
	return rt
}

func (this *Request) PassHeaders(resp *http.Response,ctx *fasthttp.RequestCtx){
	for hk,hv := range resp.Header {
		ctx.Response.Header.Set(hk,strings.Join(hv,";"))
	}
}

func readHeader(header string) [][]byte {
	reader := strings.NewReader(header)
	scaner := bufio.NewScanner(reader)
	//scaner.Scan()
	var t [][]byte
	//var s [][]byte
	for scaner.Scan() {
		t = append(t,scaner.Bytes())
		//log.Success(scaner.Text())
	}
	//copy(s,t)
	return t
}

func min(x, y int) int {
	if x < y {
		return x
	}
	return y
}

func getCircuitBreaker(service string) *circuitbreaker{
	cache := lib.NewCache()
	cb,err := cache.GetCircuitBreakerSetting(service)
	rcb :=  &circuitbreaker{
		Timeout: 3600,
		Rate: 50,
		Window: 30,
		Min: 5,
	}
	if err != nil{
		return rcb
	}
	jsoniter.ConfigCompatibleWithStandardLibrary.Unmarshal([]byte(cb),rcb)
	return rcb
}